package com.example.hbasespringbootdemo.service.impl;

import com.example.hbasespringbootdemo.constract.TableConstract;
import com.example.hbasespringbootdemo.dao.CityRowMapper;
import com.example.hbasespringbootdemo.model.City;
import com.example.hbasespringbootdemo.service.CityService;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

// HbaseTemplate 使用类似 RedisTemplate 进行api调用
@Service
public class CityServiceImpl implements CityService {

    @Autowired
    private HbaseTemplate hbaseTemplate;

    //根据开始、结束行
    @Override
    public List<City> query(String startRow, String stopRow) {
        Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
        scan.setCaching(5000);
        List<City> dtos = this.hbaseTemplate.find(TableConstract.TABLE_NAME, scan, new CityRowMapper());
        return dtos;
    }
    //查询
    @Override
    public City query(String row) {
        City dto = this.hbaseTemplate.get(TableConstract.TABLE_NAME, row, new CityRowMapper());
        return dto;
    }
    //新增保存
    @Override
    public void saveOrUpdate(String row) throws IOException {
        List<Mutation> saveOrUpdates = new ArrayList<Mutation>();
        Put put =new Put(Bytes.toBytes(row));

        put.addColumn(Bytes.toBytes("cityinfo"), Bytes.toBytes("id"), Bytes.toBytes("1"));
        put.addColumn(Bytes.toBytes("cityinfo"), Bytes.toBytes("name"), Bytes.toBytes("youxuehu"));
        put.addColumn(Bytes.toBytes("cityinfo"), Bytes.toBytes("age"), Bytes.toBytes("20"));
        saveOrUpdates.add(put);
        Connection connection = hbaseTemplate.getConnection();
        Admin admin = connection.getAdmin();

        if (!admin.tableExists(TableName.valueOf(TableConstract.CITY))) {
            HTableDescriptor table = new HTableDescriptor(TableName.valueOf(TableConstract.CITY));
            String[] family = {"cityinfo"};
            for (String str : family) {
                HColumnDescriptor column = new HColumnDescriptor(str);
                table.addFamily(column);
            }
            admin.createTable(table);
        }

        this.hbaseTemplate.saveOrUpdates(TableConstract.CITY, saveOrUpdates);
    }
}
